package tom.vertx.core.demo07;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.spi.cluster.zookeeper.ZookeeperClusterManager;
import lombok.extern.slf4j.Slf4j;

import java.time.Instant;

/**
 * shareData可以在同一个应用、同一个vertx实例多个verticle之间，集群vertx实例间共享数据(集群依赖cluster manager)
 *    localXXX: 同应用多verticle实例间共享
 *    asyncXXX: 集群内多vertx实例节点间数据共享
 *@implNote
 *    集群AsyncMap共享数据、本地AsyncMap
 * @author ZHUFEIFEI
 */
@Slf4j
public class ShareData02AsyncMapDemo {

  public static void main(String[] args) {
    //本地zk
    Vertx.clusteredVertx(new VertxOptions().setClusterManager(new ZookeeperClusterManager()), res -> {
      if (res.succeeded()) {
        res.result().deployVerticle(new V1());
        log.error("vertx cluster started.");
      } else {
        log.error("vertx cluster failed.");
      }
    });
  }

  static class V1 extends AbstractVerticle {
    @Override
    public void start(Promise<Void> startPromise) throws Exception {

      vertx.setTimer(3000L, id -> {
        vertx.sharedData().getAsyncMap("myAsyncMap", res -> {
          if (res.succeeded()) {
            log.info("getAsyncMap succeed.");
            //10秒过期时间, 再次执行操作时，key相同的value总会被更新
            // 但是集群环境下，第一个设置ttl值的会随后启动一个定时器，到达ttl时间后删除zk上的值，所以后面即使其他实例更新了ttl值也没用
            res.result().putIfAbsent("hello", "world" + Instant.now(), 10000L, comp -> {
              if (comp.succeeded()) {
                log.info("AsyncMap put succeed");
              } else {
                log.info("AsyncMap put failed => {}", comp.result());
              }
            });
          }
        });
      });


      vertx.setPeriodic(1000L, res -> {
        vertx.sharedData().getAsyncMap("myAsyncMap", rs -> {
          if(rs.succeeded()) {
            //从zk节点上获取数据
            rs.result().get("hello", ar -> {
              if (ar.succeeded()) {
                log.info("get value => {}", ar.result());
              } else {
                log.info("get value failed => {}", ar.cause().toString());
              }
            });
          } else {
            log.info("getAsyncMap value failed => {}", rs.cause().toString());
          }
        });
      });

      //LocalAsyncMap 异步Map与LocalMap不是同一个，本地共享数据的异步Map
      vertx.sharedData().getLocalAsyncMap("myLocalAsyncMap", res -> {
        if (res.succeeded()) {
          res.result().put("async", "asyncValue", cmp -> {
            if (cmp.succeeded()) {
              log.info("put async map succeed.");
            } else {
              log.info("put async map failed.");
            }
          });
        } else {
          log.info("getLocalAsyncMap error => {}", res.cause().toString());
        }
      });

      startPromise.complete();
    }
  }

}
